#pragma once

#include <Message.pb.h>
#include <codec/codec.h>
#include <codec/dispatcher.h>
#include <mq_proto.pb.h>
#include <muduo/net/TcpConnection.h>

#include <Common/thread_pool.hpp>
#include <MQ_Virtual_Engine/VirtualHost.hpp>
#include <cstddef>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <string_view>
#include <unordered_map>
#include <utility>

#include "Comsumer.hpp"
#include "MQ_Virtual_Engine/BindInfo.hpp"
#include "muduo/net/Callbacks.h"
#include "router.hpp"

namespace MY_MQ {
using ProtobufCodecPtr          = std::shared_ptr<ProtobufCodec>;
using openChannelRequestPtr     = std::shared_ptr<MQ_Protobuf::openChannelRequest>;
using closeChannelRequestPtr    = std::shared_ptr<MQ_Protobuf::closeChannelRequest>;
using declareExchangeRequestPtr = std::shared_ptr<MQ_Protobuf::declareExchangeRequest>;
using deleteExchangeRequestPtr  = std::shared_ptr<MQ_Protobuf::deleteExchangeRequest>;
using declareQueueRequestPtr    = std::shared_ptr<MQ_Protobuf::declareQueueRequest>;
using deleteQueueRequestPtr     = std::shared_ptr<MQ_Protobuf::deleteQueueRequest>;
using queueBindRequestPtr       = std::shared_ptr<MQ_Protobuf::queueBindRequest>;
using queueUnBindRequestPtr     = std::shared_ptr<MQ_Protobuf::queueUnBindRequest>;
using basicPublishRequestPtr    = std::shared_ptr<MQ_Protobuf::basicPublishRequest>;
using basicAckRequestPtr        = std::shared_ptr<MQ_Protobuf::basicAckRequest>;
using basicConsumeRequestPtr    = std::shared_ptr<MQ_Protobuf::basicConsumeRequest>;
using basicCancelRequestPtr     = std::shared_ptr<MQ_Protobuf::basicCancelRequest>;

class Channel {
    std::string                      _ChannelID;
    Consumer::self_ptr               _ConsumerPtr;
    muduo::net::TcpConnectionPtr     _conn;
    ProtobufCodecPtr                 _codec;
    ConsumerManager::self_ptr        _ConsumperMnagePtr;
    VirtualHost::self_ptr            _VirtualHostPtr;
    Message_MQ::ThreadPool::self_ptr _threadpool;

public:
    using self_ptr = std::shared_ptr<Channel>;

    Channel(const std::string &ChannelID, const VirtualHost::self_ptr &host, const ProtobufCodecPtr &codec,
            const muduo::net::TcpConnectionPtr &conn,
            const ConsumerManager::self_ptr &consumerManagerPtr, const Message_MQ::ThreadPool::self_ptr &threadpool)
        : _ChannelID(ChannelID),
          _conn(conn),
          _codec(codec),
          _ConsumperMnagePtr(consumerManagerPtr),
          _VirtualHostPtr(host),
          _threadpool(threadpool) {
    }

    ~Channel() {
        if (nullptr != _ConsumerPtr) {
            _ConsumperMnagePtr->removeConsumerFromQueue(_ConsumerPtr->_SubSriber_QueueName, _ConsumerPtr->_ComsumerID);
        }
        logd("del channel,{}", (void *)this);
    }

    // 声明交换机和删除交换机
    void DeclareExchange(const declareExchangeRequestPtr &req) {
        bool ret = _VirtualHostPtr->DeclareExchange(req->exchange_name(), req->exchange_type(), req->durable(),
                                                    req->auto_delete(), req->args());
        return basicResponse(ret, req->rid(), req->cid());
    }

    void deleteExchange(const deleteExchangeRequestPtr &req) {
        return basicResponse(_VirtualHostPtr->DeleteExchange(req->exchange_name()), req->rid(), req->cid());
    }

    void DeclareQueue(const declareQueueRequestPtr &req) {
        auto res = _VirtualHostPtr->DeclareQueue(req->queue_name(), req->durable(), req->exclusive(),
                                                 req->auto_delete(), req->args());
        if (!res) {
            return basicResponse(false, req->rid(), req->cid());
        }
        _ConsumperMnagePtr->InitComsumerQueue(req->queue_name());
        return basicResponse(true, req->rid(), req->cid());
    }

    void DeleteQueue(const deleteQueueRequestPtr &req) {
        _ConsumperMnagePtr->RemoveComsumerQueue(req->queue_name());
        _VirtualHostPtr->DeleteQueue(req->queue_name());
        return basicResponse(true, req->rid(), req->cid());
    }

    void queueBind(const queueBindRequestPtr &req) {
        return basicResponse(_VirtualHostPtr->Bind(req->exchange_name(), req->queue_name(), req->binding_key()),
                             req->rid(), req->cid());
    }

    void queueUbind(const queueUnBindRequestPtr &req) {
        _VirtualHostPtr->UnBind(req->exchange_name(), req->queue_name());
        return basicResponse(1, req->rid(), req->cid());
    }

    void basicPublish(const basicPublishRequestPtr &req) {
        // 1.first to do is exchange is exsits
        auto exchange_ptr = _VirtualHostPtr->selectExchange(req->exchange_name());
        if (nullptr == exchange_ptr) {
            return basicResponse(false, req->rid(), req->cid());
        }
        // 2.select binding's exchange_machine by routingkey
        auto                          mqbm       = _VirtualHostPtr->ExchangeBindMap(req->exchange_name());
        MQ_Protobuf::BasicProperties *properties = nullptr;
        std::string                   routing_key;
        if (req->has_properties()) {
            properties  = req->mutable_properties();
            routing_key = properties->routing_key();
        }
        for (auto &binding : mqbm) {
            if (Router::route(exchange_ptr->_type, routing_key, binding.second->_BindingKey)) {
                if (properties)
                    _VirtualHostPtr->BasicPublish(binding.first, *properties, req->body());
                else {
                    _VirtualHostPtr->BasicPublish(binding.first, req->body());
                };
                auto task = std::bind(&Channel::consume, this, binding.first);
                _threadpool->enqueue(std::move(task));
            }
        }
        return basicResponse(true, req->rid(), req->cid());
    }

    void basicAck(const basicAckRequestPtr &req) {
        _VirtualHostPtr->BasicAck(req->queue_name(), req->message_id());
        return basicResponse(1, req->rid(), req->cid());
    }

    void basicConsume(const basicConsumeRequestPtr &req) {
        bool res = _VirtualHostPtr->ExsitQueue(req->queue_name());
        if (false == res) {
            logd("队列{}不存在",req->queue_name());
            return basicResponse(false, req->rid(), req->cid());
        }
        auto cb =
            std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
        _ConsumerPtr = _ConsumperMnagePtr->Create(req->consumer_tag(), req->queue_name(), req->auto_ack(), cb);
        return basicResponse(1, req->rid(), req->cid());
    }

    void basicCancel(const basicCancelRequestPtr &req) {
        _ConsumperMnagePtr->removeConsumerFromQueue(req->consumer_tag(), req->queue_name());
        return basicResponse(1, req->rid(), req->cid());
    }

private:
    void callback(const std::string Tag, const MQ_Protobuf::BasicProperties *bp, const std::string &body) {
        MQ_Protobuf::basicConsumeResponse resp;
        resp.set_cid(_ChannelID);
        resp.set_body(body);
        resp.set_consumer_tag(Tag);
        if (bp) {
            resp.mutable_properties()->set_id(bp->id());
            resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
            resp.mutable_properties()->set_routing_key(bp->routing_key());
        }
        _codec->send(_conn, resp);
    }

    void consume(const std::string &qname) {
        MessagePtr         l_mp = _VirtualHostPtr->BasicConsum(qname);
        if (nullptr == l_mp) {
            logd("消费任务执行失败，{}队列没有消息", qname);
            return ;
        }
        Consumer::self_ptr l_cp = _ConsumperMnagePtr->ChooseByTurn(qname);
        if (nullptr == l_cp) {
            logd("消费任务执行失败，{}队列没有消费者", qname);
            return ;
        }
        l_cp->_call(l_cp->_ComsumerID, l_mp->mutable_payload()->mutable_properties(), l_mp->payload().body());
        if (l_cp->_AutoAck) {
            _VirtualHostPtr->BasicAck(qname, l_mp->payload().properties().id());
        }
    }

    void basicResponse(bool ok, const std::string &rid, const std::string &cid) {
        MQ_Protobuf::basicCommonResponse resp;
        resp.set_rid(rid);
        resp.set_cid(cid);
        resp.set_ok(ok);
        _codec->send(_conn, resp);
    }
};

class ChannelManager {
public:
    using self_ptr = std::shared_ptr<ChannelManager>;

    bool OpenChannel(const std::string &id, const VirtualHost::self_ptr &host, const ConsumerManager::self_ptr &cmp,
                     const Message_MQ::ThreadPool::self_ptr &tp, const ProtobufCodecPtr &codec,
                     const muduo::net::TcpConnectionPtr &conn) {
        std::lock_guard lk(_mutex);
        auto            it = _channels.find(id);
        if (it != _channels.end()) {
            return false;
        }
        _channels[id] = std::make_shared<Channel>(id, host, codec,  conn,cmp,tp );
        return true;
    }

    void CloseChannel(const std::string &id) {
        std::lock_guard lk(_mutex);
        auto            it = _channels.find(id);
        if (it != _channels.end()) {
            _channels.erase(it);
        }
    }

    Channel::self_ptr getChannel(const std::string &id) {
        std::lock_guard lk(_mutex);
        auto            it = _channels.find(id);
        if (it != _channels.end()) {
            return it->second;
        }
        return nullptr;
    }

private:
    std::mutex                                         _mutex;
    std::unordered_map<std::string, Channel::self_ptr> _channels;
};
};  // namespace MY_MQ